package org.example.server;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import java.math.BigDecimal;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;

/**
 * @author: ReWinD00
 * @date: 2021-07-05 10:43
 */
@Component
@Scope("prototype")
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
	@Autowired
    private RocketMQTemplate rocketMQTemplate;
	@Autowired
    private ObjectMapper objectMapper;
	
	private int idleCounter = 0;
	DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

	/**
	 * 业务数据处理
	 *
	 * @param ctx
	 * @param msg
	 * @throws Exception
	 */
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		try {
			idleCounter = 0;
			
			ByteBuf in = (ByteBuf) msg;
			int readableBytes = in.readableBytes();
			byte[] bytes = new byte[readableBytes];
			in.readBytes(bytes);
			
			String clientIp = this.getClientIp(ctx);
			String msgStr = new String(bytes, StandardCharsets.UTF_8);
			log.info("收到{}数据{} ", clientIp, msgStr);
			if(StringUtils.isBlank(msgStr)) {
        		return;
        	}
			
			String[] msgArr = msgStr.split(",");
			List<Map<String, Object>> data = Lists.newArrayList();
			Map<String, Object> dataMap = Maps.newHashMap();
			dataMap.put("id", "x");
			dataMap.put("key", "x");
			dataMap.put("val", new BigDecimal(msgArr[1]));
			data.add(dataMap);
			
			dataMap = Maps.newHashMap();
			dataMap.put("id", "y");
			dataMap.put("key", "y");
			dataMap.put("val", new BigDecimal(msgArr[2]));
			data.add(dataMap);
			
			dataMap = Maps.newHashMap();
			String msg3 = msgArr[3];
			dataMap.put("id", "h");
			dataMap.put("key", "h");
			// 解决粘包问题，该数据截取小数点后2位，其他数据丢弃
			dataMap.put("val", new BigDecimal(msg3.substring(0, msg3.indexOf(".") + 3)));
			data.add(dataMap);
			
			// 组装采集器数据
			Map<String, Object> deviceMap = Maps.newHashMap();
			deviceMap.put("devId", msgArr[0]);
			deviceMap.put("sn", msgArr[0]);
			deviceMap.put("time", dtf.format(LocalDateTime.now()));
			deviceMap.put("tenantId", "");
			deviceMap.put("tags", null);
			deviceMap.put("data", data);
			
			String message = objectMapper.writeValueAsString(deviceMap);
			
			rocketMQTemplate.asyncSend("iot-huacheng", MessageBuilder.withPayload(message).setHeader("KEYS", msgArr[0]).build(), new SendCallback() {
				@Override
                public void onSuccess(SendResult result) {
                    if (result.getSendStatus() == SendStatus.SEND_OK) {
                    	log.info("{}数据发送成功={}", clientIp, message);
                    }
                }
                @Override
                public void onException(Throwable e) {
                	log.error("{}数据发送失败={}", clientIp, e.getMessage());
                }
            });
		} finally {
			// 抛弃收到的数据
			ReferenceCountUtil.release(msg);
		}
	}

	/**
	 * 从客户端收到新的数据、读取完成---调用
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		log.info("{}数据读取完成", this.getClientIp(ctx));
		if (ctx != null) {
			ctx.flush();
		}
	}

	/**
	 * 客户端与服务端建立连接--执行
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
		ctx.channel().read();
		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
		String clientIp = insocket.getAddress().getHostAddress();
		// 此处不能使用ctx.close()，否则客户端始终无法与服务端建立连接
		log.info("{}上线", clientIp);
	}

	/**
	 * 客户端与服务端断连-调用
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);
		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
		String clientIp = insocket.getAddress().getHostAddress();
		// 断开连接时，服务端关闭连接，避免造成资源浪费
		ctx.close();
		log.info("{}下线", clientIp);
	}

	/**
	 * 当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常
	 *
	 * @param ctx
	 * @param cause
	 * @throws Exception
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		if (ctx != null) {
			ChannelHandlerContextHolder.removeByCtx(ctx);
			// 抛出异常，断开与客户端的连接
			ctx.close();
			log.error("连接异常，服务端主动断开连接{}", cause.getMessage(), cause);
		}
	}

	/**
	 * 服务端read超时-调用
	 *
	 * @param ctx
	 * @param evt
	 * @throws Exception
	 */
	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

		String clientIp = this.getClientIp(ctx);

		if (evt instanceof IdleStateEvent) {
			IdleStateEvent e = (IdleStateEvent) evt;
			if (e.state().equals(IdleState.WRITER_IDLE)) {
				log.warn("{}写超时", clientIp);
				idleCounter++;
				if (idleCounter > 3) {
					log.warn("{}写超时超过3次，断开", clientIp);
					ctx.disconnect();
				}
			}
		}
	}

	private String getClientIp(ChannelHandlerContext ctx) {
		InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
		String clientIP = insocket.getAddress().getHostAddress();
		return clientIP;
	}
	
	public static void main(String[] args) {
		// 粘包
		String data = "X22CPKOA0078,0.51,0.25,-0.52X22CPKOA0078,0.51,0.25,0.52X22CPKOA0078,0.51,0.25,0.52";
		String[] msgArr = data.split(",");
		String msg3 = msgArr[3];
		System.out.println(msg3.substring(0, msg3.indexOf(".") + 3));
		System.out.println(msg3.substring(msg3.indexOf(".") + 3));
	}
}